package com.nifostasky.mqcallback.consumer;

/**
 *  消费者
 */

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * ## 消息的可靠消费
 */
@Slf4j
@Component
public class CallbackConsumer {

    /**
     * 消费信息的三种回执方式
     * basicAck:
     *      方法:
     *          void basicAck(long deliveryTag, boolean multiple) throws IOException;
     *      参数:
     *          deliveryTag: 消息投递的标签号,每次消费消息或者消息重新投递后，deliveryTag都会增加。手动消息确认模式下，我们可以对指定deliveryTag的消息进行ack、nack、reject等操作。
     *          multiple: 是否批量确认,值为 true 则会一次性 ack所有小于当前消息 deliveryTag 的消息
     * basicNack:
     *      方法:
     *          void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;
     *      参数:
     *          deliveryTag：表示消息投递序号。
     *          multiple：是否批量确认。
     *          requeue：值为 true 消息将重新入队列。
     * basicReject:
     *      方法:
     *          void basicReject(long deliveryTag, boolean requeue) throws IOException;
     *      参数:
     *          deliveryTag：表示消息投递序号。
     *          requeue：值为 true 消息将重新入队列。
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = {"helloRabbitMQ"})
    public void process(Message message, Channel channel) throws IOException {
        MessageProperties messageProperties = message.getMessageProperties();
        log.info(messageProperties.toString());
        try {
            log.info(message.toString());
            log.info(new String(message.getBody()));
            int a = 1/0;
            //确认消息
            channel.basicAck(messageProperties.getDeliveryTag(), false);
        } catch (Exception e) {
            // 当前的消息是否重新投递的消息,也就是该消息是重新回到队列里的消息
            if (messageProperties.getRedelivered()) {
                log.info("消息已重复处理失败,拒绝再次接收...");
                // 拒绝消息
                channel.basicReject(messageProperties.getDeliveryTag(), false);
            } else {
                log.info("消息即将再次返回队列处理...");
                //重新处理
                channel.basicNack(messageProperties.getDeliveryTag(), false, true);
            }
//            解决 消息无限消费
//            // 重新发送消息到队尾
//            channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
//                    message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
//                    JSON.toJSONBytes(msg));
        }
    }

}